package org.hughie.core;


import org.hughie.common.MConstant;
import org.hughie.common.SyncConfig;
import org.hughie.log.Logger;
import org.hughie.log.LoggerFactory;

import javax.swing.*;
import java.sql.*;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class DataSyncEngine {

    static Logger logger = LoggerFactory.getLogger(DataSyncEngine.class);

    private static final int BATCH_SIZE = 1000;

    public static void syncData(JFrame parent, SyncConfig config) {
        long start = System.currentTimeMillis();
        // 加载类驱动
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException ex) {
            JOptionPane.showMessageDialog(parent, "类加载失败！" + ex.getMessage(), "ERROR", JOptionPane.ERROR_MESSAGE);
            ex.printStackTrace();
        }
        ExecutorService executor = Executors.newFixedThreadPool(config.getThreadCount());
        int totalRows;
        try (Connection sourceConn = DriverManager.getConnection(config.getSourceDb());
             Connection targetConn = DriverManager.getConnection(config.getTargetDb());
             Statement countStmt = sourceConn.createStatement();
             Statement targetStmt = targetConn.createStatement()) {
            sourceConn.setAutoCommit(false);
            targetConn.setAutoCommit(false);
            // 删除目标表数据
            String delSQL = "DELETE FROM " + config.getTableName() + " WHERE " + config.getCondition();
            if (MConstant.CONDITION_1_1.equals(config.getCondition())) {
                delSQL = "truncate table " + config.getTableName();
            }
            targetStmt.executeUpdate(delSQL);
            targetConn.commit();
            ArrayList<Future> futures = new ArrayList<>();
            //分批插入目标表数据
            try (ResultSet countRs = countStmt.executeQuery("SELECT COUNT(*) FROM " + config.getTableName() + " WHERE " + config.getCondition())) {
                countRs.next();
                totalRows = countRs.getInt(1);
                int rowsPerThread = (int) Math.ceil((double) totalRows / config.getThreadCount());
                for (int i = 0; i < config.getThreadCount(); i++) {
                    int offset = i * rowsPerThread;
                    futures.add(executor.submit(() -> {
                        try {
                            fetchAndInsert(sourceConn, targetConn, config, offset, rowsPerThread);
                        } catch (Exception ex) {
                            JOptionPane.showMessageDialog(parent, "数据同步失败！" + ex.getMessage(), "ERROR", JOptionPane.ERROR_MESSAGE);
                            ex.printStackTrace();
                        }
                    }));
                }
            }
            futures.forEach(t -> {
                try {
                    t.get();
                } catch (Exception ex) {
                    JOptionPane.showMessageDialog(parent, "数据同步失败！" + ex.getMessage(), "ERROR", JOptionPane.ERROR_MESSAGE);
                    ex.printStackTrace();
                }
            });
            JOptionPane.showMessageDialog(parent, "数据同步成功！\n数据" + totalRows + "条，耗时：" + (float) ((System.currentTimeMillis() - start) / 1000) + "s");
        } catch (Exception ex) {
            ex.printStackTrace();
            JOptionPane.showMessageDialog(parent, "数据同步失败！" + ex.getMessage(), "ERROR", JOptionPane.ERROR_MESSAGE);
        }
        executor.shutdown();
    }

    private static void fetchAndInsert(Connection sourceConn, Connection targetConn, SyncConfig config,
                                       int offset, int limit) throws Exception {
        String selectQuery = "SELECT * FROM " + config.getTableName() + " WHERE " + config.getCondition() + " LIMIT " + limit + " OFFSET " + offset;
        long start1 = System.currentTimeMillis();
        try (Statement stmt = sourceConn.createStatement();) {
            stmt.setFetchSize(BATCH_SIZE);
            ResultSet rs = stmt.executeQuery(selectQuery);
            ResultSetMetaData metaData = rs.getMetaData();
            int columnCount = metaData.getColumnCount();
            logger.info("查询耗时：{}ms", System.currentTimeMillis() - start1);
            StringBuilder insertQuery = new StringBuilder("REPLACE INTO " + config.getTableName() + " VALUES (");
            for (int i = 0; i < columnCount - 1; i++) {
                insertQuery.append("?, ");
            }
            insertQuery.append("?)");
            try (PreparedStatement pstmt = targetConn.prepareStatement(insertQuery.toString())) {
                int batchSize = 0;
                long startN = System.currentTimeMillis();
                long startM;
                while (rs.next()) {
                    for (int i = 1; i <= columnCount; i++) {
                        Object object = rs.getObject(i);
                        startM = System.currentTimeMillis();
                        int columnType = metaData.getColumnType(i);
                        if (null == object) {
                            pstmt.setNull(i, columnType);
                            if ((System.currentTimeMillis() - startM) > 100) {
                                logger.info("batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                            }
                            continue;
                        }
                        switch (columnType) {
                            case Types.INTEGER:
                                pstmt.setInt(i, rs.getInt(i));
                                if ((System.currentTimeMillis() - startM) > 100) {
                                    logger.info("{} batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", "INTEGER", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                                }
                                break;
                            case Types.VARCHAR:
                                pstmt.setString(i, rs.getString(i));
                                if ((System.currentTimeMillis() - startM) > 100) {
                                    logger.info("{} batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", "VARCHAR", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                                }
                                break;
                            case Types.DOUBLE:
                                pstmt.setDouble(i, rs.getDouble(i));
                                if ((System.currentTimeMillis() - startM) > 100) {
                                    logger.info("{} batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", "DOUBLE", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                                }
                                break;
                            case Types.TIMESTAMP:
                                pstmt.setTimestamp(i, rs.getTimestamp(i));
                                if ((System.currentTimeMillis() - startM) > 100) {
                                    logger.info("{} batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", "TIMESTAMP", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                                }
                                break;
                            default:
                                pstmt.setObject(i, rs.getObject(i)); // 只对无法明确类型的数据使用
                                break;
                        }
                        if ((System.currentTimeMillis() - startM) > 100) {
                            logger.info("{} batchSize:{} setObject长耗时：{}ms for-i: {} id:{} colName:{},colValue:{}", "Others", batchSize, (System.currentTimeMillis() - startM), i, rs.getObject("id"), metaData.getColumnName(i), object);
                        }
                    }
                    pstmt.addBatch();
                    batchSize++;
                    //达到批数量即提交一次
                    if (batchSize % BATCH_SIZE == 0) {
                        logger.info("batchSize:" + batchSize + " 1.构建批数据耗时：" + (System.currentTimeMillis() - startN) + "ms");
                        startN = System.currentTimeMillis();
                        pstmt.executeBatch();
                        targetConn.commit();
                        pstmt.clearBatch();
                        logger.info("batchSize:" + batchSize + " 2.插入数据" + BATCH_SIZE + "条耗时：" + (System.currentTimeMillis() - startN) + "ms");
                        startN = System.currentTimeMillis();
                    }
                }
                pstmt.executeBatch();
                targetConn.commit();
                rs.close();
                logger.info("3.插入数据" + batchSize % BATCH_SIZE + "条耗时：" + (System.currentTimeMillis() - startN) + "ms");
            }
        }
    }
}
